#pragma once

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <functional>
#include <string>
#include <thread>

#include "logger.hpp"

namespace xu
{
    class MQClient
    {
    public:
        using ptr = std::shared_ptr<MQClient>;
        using MessageCallback = std::function<void(const char *, uint64_t)>;
        MQClient(const std::string &user, const std::string &password, const std::string &host)
        {
            _loop = EV_DEFAULT;
            _ev_handler = std::make_unique<AMQP::LibEvHandler>(_loop);
            //"amqp://root:xzyxzy99@127.0.0.1:5672/"
            std::string addr = "amqp://" + user + ":" + password + "@" + host + "/";
            AMQP::Address address(addr);
            _conn = std::make_unique<AMQP::TcpConnection>(_ev_handler.get(), address);
            _channel = std::make_unique<AMQP::TcpChannel>(_conn.get());

            _thread = std::make_unique<std::thread>([this]()
                                                    { ev_run(_loop, 0); });
        }

        void DeclareComponents(const std::string &exchange_name, const std::string &queue_name, const std::string &routing_key,
                               AMQP::ExchangeType etype = AMQP::ExchangeType::direct)
        {
            _channel->declareExchange(exchange_name, etype)
                .onError([](const char *message)
                         {
            LOG_ERROR("声明交换机失败：{}",message);
            exit(0); })
                .onSuccess([]()
                           { LOG_INFO("交换机创建成功！"); });

            // 5. 声明队列
            _channel->declareQueue(queue_name)
                .onError([](const char *message)
                         {
            LOG_ERROR("声明队列失败：{}",message);
            exit(0); })
                .onSuccess([]()
                           { LOG_INFO("队列创建成功！"); });
            // 6. 针对交换机和队列进行绑定
            _channel->bindQueue(exchange_name, queue_name, routing_key)
                .onError([exchange_name, queue_name](const char *message)
                         { 
                            LOG_ERROR("{} - {}绑定失败：{}",exchange_name,queue_name,message);
                            exit(0); })
                .onSuccess([]()
                           { LOG_INFO("绑定成功！"); });
        }

        bool Publish(const std::string &exchange_name, const std::string &routing_key,
                     const std::string &msg)
        {
            bool ret = _channel->publish(exchange_name, routing_key, msg);
            return ret;
        }
        void Consume(const std::string &queu_name, const std::string &tag, MessageCallback msg_callback)
        {
            auto callback = std::bind(&MQClient::MessageCb, this, msg_callback, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            _channel->consume(queu_name, tag) // 返回值 DeferredConsumer
                .onReceived(callback)
                .onError([queu_name](const char *message)
                         {
            std::cout << "订阅 test-queue 队列消息失败:" << message << std::endl;
            LOG_ERROR("订阅 {}队列消息失败:{}",queu_name,message);
            exit(0); }); // 返回值是 AMQP::Deferred
        }
        ~MQClient()
        {
            // void ev_async_init(ev_async *w, callback cb);
            // void ev_async_start(struct ev_loop *loop, ev_async *w) ;
            // void ev_async_send(struct ev_loop *loop, ev_async *w) ;
            ev_async_init(&_ev_async, EVCallback);
            ev_async_start(_loop, &_ev_async);
            ev_async_send(_loop, &_ev_async);
            _thread->join();
            _loop = nullptr;
        }

    private:
        void MessageCb(MessageCallback msg_callback, const AMQP::Message &message,
                       uint64_t deliveryTag, bool redelivered)
        {
            if (msg_callback)
                msg_callback(message.body(), message.bodySize());

            _channel->ack(deliveryTag);
        }

        static void EVCallback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
        {
            ev_break(loop, EVBREAK_ALL);
        }

    private:
        struct ev_loop *_loop;
        struct ev_async _ev_async;
        std::unique_ptr<AMQP::LibEvHandler> _ev_handler;
        std::unique_ptr<AMQP::TcpConnection> _conn;
        std::unique_ptr<AMQP::TcpChannel> _channel;
        std::unique_ptr<std::thread> _thread;
    };

}